package cn.itcast.function;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.IndexBean;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class SecIndexWindowFunction extends RichWindowFunction<CleanBean, IndexBean, String, TimeWindow> {
    @Override
    public void apply(String key, TimeWindow window, Iterable<CleanBean> input, Collector<IndexBean> out) throws Exception {
//        1.新建SecIndexWindowFunction 窗口函数
//2.记录最新指数
        CleanBean line = null;
        for (CleanBean cleanBean : input) {
            if (line == null) {
                line = cleanBean;
            }
            if (cleanBean.getEventTime() > line.getEventTime()) {
                line = cleanBean;
            }
//3.格式化日期
            Long aLong = DateUtil.longTimeTransfer(line.getEventTime(), Constant.format_yyyymmdd);
//4.封装输出数据
            out.collect(new IndexBean(

                            cleanBean.getEventTime(),
                            cleanBean.getSecCode(),
                            cleanBean.getSecName(),
                            cleanBean.getPreClosePrice(),
                            cleanBean.getOpenPrice(),
                            cleanBean.getMaxPrice(),
                            cleanBean.getMinPrice(),
                            cleanBean.getTradePrice(),
                            0l,0l,
                            cleanBean.getTradeVolumn(),
                            cleanBean.getTradeAmt(),
                            aLong,
                            cleanBean.getSource()
            )
            );
        }
    }
}
